package com.flink.timewindow.window;

import com.flink.timewindow.bean.WaterSensor;
import com.flink.timewindow.function.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class WindowAggregateAndProcessDemo {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        //获取数据源
        SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("10.90.100.102", 8888)
                //数据处理
                //切分转换
                .map(new WaterSensorMapFunction());
        //分组
        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(value -> value.getId());

        //窗口分配器
        WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorKS.window(
                TumblingProcessingTimeWindows.of(Time.seconds(10))
        );


        //窗口函数
        /*
        增量聚台 Aggregate+全窗口 process
        ①、增量聚合函数处理数据:来一条计算一条
        ②窗口触发时，增量聚合的结果(只有一条)传递给全窗口函数
        ③经过全窗口函数的处理包装后，输出
         结合两者的优点:
         ①增量聚合: 来一条计算一条，存储中间的计算结果，占用的空间少
         ②全窗口函数:可以通过 上下文 实现灵活的功能
        *
        * */
        SingleOutputStreamOperator<String> result = sensorWS.aggregate(new CustomAggregate(), new CustomProcess());
        //输出
        result.print();


        //执行
        env.execute();

    }
    //定义实现AggregateFunction接口
    public static class CustomAggregate implements AggregateFunction<WaterSensor, Integer, String>{

        @Override
        public Integer createAccumulator() {//创建累加器，即初始化累加器
            System.out.println("创建累加器");
            return 0;
        }

        @Override
        public Integer add(WaterSensor value, Integer accumulator) {//将输入的元素添加到累加器中，聚合逻辑
            System.out.println("调用add方法，value：" + value + " ,accumulator：" + accumulator);

            return accumulator + value.getVc();
        }

        @Override
        public String getResult(Integer accumulator) {//从累加器中提取聚合的输出结果即获取最终结果，窗口触发时输出
            System.out.println("调用getResult方法");
            return accumulator.toString();
        }

        @Override
        public Integer merge(Integer a, Integer b) {//合并两个累加器，并将合并后的状态作为一个累加器返回,但这只有会话窗口才会用到
            System.out.println("调用merge方法");

            return 0;
        }
    }

    //定义ProcessWindowFunction实现类 ,需要注意的是这里的输入类型是Aggregate的输出类型
    public static class CustomProcess extends ProcessWindowFunction<String, String, String, TimeWindow> {

        @Override
        public void process(String s,
                            ProcessWindowFunction<String, String, String, TimeWindow>.Context context,
                            Iterable<String> elements,
                            Collector<String> out) throws Exception {
            //上下文可以拿到window对象，还有其他东西：如测流输出
            long count = elements.spliterator().estimateSize();
            long windowStartTs = context.window().getStart();
            long windowEndTs = context.window().getEnd();
            String windowStart = DateFormatUtils.format(windowStartTs, "yyyy-MM-dd HH:mm:ss.SSS");
            String windowEnd = DateFormatUtils.format(windowEndTs, "yyyy-MM-dd HH:mm:ss.SSS");

            out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());

        }
    }

}
